package com.jhhe.homework4_2

import org.apache.spark.rdd.RDD
import org.apache.spark.{SPARK_BUILD_USER, SparkConf, SparkContext}
/**
 * @Description:
 * 需求:  将http.log文件中的ip转换为地址。如将 122.228.96.111 转为温州，并统计各城市的总访问量
 * http.log数据格式：时间戳、IP地址、访问网址、访问数据、浏览器信息
 * http.log数据样例：20090121000132095572000|125.213.100.123|show.51.com|/shoplist.php?phpfile=shoplist2.php&style=1&sex=137|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; Mozilla/4.0(Compatible Mozilla/4.0(Compatible-EmbeddedWB 14.59 http://bsalsa.com/ EmbeddedWB- 14.59  from: http://bsalsa.com/ )|http://show.51.com/main.php|
 *
 * @Author Jhhe
 * @Date 2021-03-30
 * @Time 15:46
 */
object Ip2Address {
  def main(args: Array[String]): Unit = {
    //1. 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    //2. 读取本地文件
    val httpLines:RDD[String] = sc.textFile("file:///E:\\MyWorks\\MyBigData\\phase-4-task-2\\data\\http.log")
    val ipLines:RDD[String] = sc.textFile("file:///E:\\MyWorks\\MyBigData\\phase-4-task-2\\data\\ip.dat")

    //3. 解析ip.dat数据：（开始IP，结束IP，网段所属城市）
    val ipAddrRDD = ipLines.map{ line =>
      val fields = line.split("\\|")
      (fields(0), fields(1), fields(7))
    }

    //4. 解析http.log数据：（ip, 一行数据）
    val httpRDD = httpLines.map { line =>
      val fields: Array[String] = line.split("\\|")
      (fields(1), line)
    }

    //5. IP去重
    val logIpRDD = httpRDD.map(_._1).distinct()

    //6. 求出日志中的ip和地址的映射数据集（ip,城市）
    val ip_addr = ipAddrRDD.cartesian(logIpRDD)
      .filter{
        case ((startIp, endIp, _), ip) => ipInRange(ip, startIp, endIp)
      }
      .map{
        case((_, _, addr), ip) => (ip, addr)
      }

    /*// 左连接：（城市，IP转为城市后的数据）
    val rdd = httpRDD.leftOuterJoin(ip_addr)
        .map {
          case (ip, (line, opt)) => (opt.get, line.replace(ip, opt.get))
        }

    // 分组，统计每组数据集大小即为各城市的总访问量
    rdd.groupByKey()
      .map{case (addr, data) => (addr, data.size)}
      .foreach(x => println(x._1 + ":" + x._2))*/
    // (城市，IP转为城市后的数据）使用广播变量，避免了shuffle
    val bc = sc.broadcast(ip_addr.collectAsMap())
    val cityRDd = httpRDD.map {
      case (ip, line) =>
        val city = bc.value.getOrElse(ip, null)
        (city, line.replace(ip, city))
    }

    // 缓存
    cityRDd.cache()

    // IP转为城市后的数据输出到文件
    cityRDd.map(_._2).saveAsTextFile("data/logout")

    sc.stop()
  }

  /*
  * 判断IP是否在开始IP和结束IP的区间内，返回Boolean
  *
  * */
  def ipInRange(IP:String, startIp: String, endIp:String): Boolean = {
    Ip2Long(startIp) <= Ip2Long(IP) && Ip2Long(IP) <= Ip2Long(endIp)
  }

  def Ip2Long(ip:String): Long = {
    val ips = ip.split("\\.") // ip.split("[.]")
    var ip2long = 0L
    for(i <- 0 until ips.length) {
      ip2long = ip2long << 8 | ips(i).toLong
    }
    ip2long
  }
}
